package com.jackstraw.jack_base.datasource.elasticsearch.util;

import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.JsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.TransportOptions;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.jackstraw.jack_base.datasource.elasticsearch.entity.bean.ElasticSearchPropertiesEntity;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.Node;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;

import javax.annotation.Nullable;
import java.time.Duration;
import java.util.List;

@Slf4j
public class ElasticSearchCreateUtil {


    public static ElasticsearchTransport elasticsearchTransport(RestClient restClient, JsonpMapper mapper, @Nullable TransportOptions options) {
        return new RestClientTransport(restClient, mapper, options);
    }

    public static ElasticsearchTransport elasticsearchTransport(RestClient restClient, JsonpMapper mapper) {
        return new RestClientTransport(restClient, mapper);
    }


    public static ElasticsearchClient elasticsearchClient(ElasticsearchTransport elasticsearchTransport) {
        return new ElasticsearchClient(elasticsearchTransport);
    }

    public static ElasticsearchClient elasticsearchClient(ElasticsearchTransport elasticsearchTransport, @Nullable TransportOptions options) {
        return new ElasticsearchClient(elasticsearchTransport, options);
    }

    public static ElasticsearchAsyncClient elasticsearchAsyncClient(ElasticsearchTransport elasticsearchTransport) {
        return new ElasticsearchAsyncClient(elasticsearchTransport);
    }

    public static RestClientBuilder restClientBuilder(ElasticSearchPropertiesEntity elasticsearchClientEntity) {
        List<String> endpoints = elasticsearchClientEntity.getEndpoints();
        String schema = elasticsearchClientEntity.getScheme();
        String userName = elasticsearchClientEntity.getUsername();
        String password = elasticsearchClientEntity.getPassword();
        Duration connectionTimeout = elasticsearchClientEntity.getConnectionTimeout();
        Duration socketTimeout = elasticsearchClientEntity.getSocketTimeout();

        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(userName, password));
        RestClientBuilder restClientBuilder = RestClient.builder(makeHttpHost(endpoints, schema));
        // 设置一个监听器，每次节点出现故障时都会收到通知，以防需要采取措施，
        // 当启用故障嗅探时在内部使用。
        restClientBuilder.setFailureListener(new RestClient.FailureListener() {
            @Override
            public void onFailure(Node node) {
                System.out.println("节点异常:" + node.toString());
            }
        });
        // 设置允许修改默认请求配置的回调
        //（例如请求超时，身份验证或org.apache.http.client.config.RequestConfig.Builder允许设置的任何内容）。
//        restClientBuilder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
//            @Override
//            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
//                return requestConfigBuilder
//                        .setSocketTimeout(socketTimeout.getNano())
//                        .setConnectTimeout(connectionTimeout.getNano())
//                        ;
//            }
//        }).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
//            @Override
//            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
//                httpClientBuilder.disableAuthCaching();
//                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
//            }
//        });
        restClientBuilder.setRequestConfigCallback((requestConfigBuilder) -> {
            return requestConfigBuilder
                    .setSocketTimeout(socketTimeout.getNano())
                    .setConnectTimeout(connectionTimeout.getNano())
                    ;
        }).setHttpClientConfigCallback((httpClientBuilder) -> {
            httpClientBuilder.disableAuthCaching();
            return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
        });
        return restClientBuilder;
    }


    /**
     * 创建HttpHost
     *
     * @param endpoints
     * @param schema
     * @return
     */
    private static HttpHost[] makeHttpHost(List<String> endpoints, String schema) {
        HttpHost[] httpHosts = new HttpHost[endpoints.size()];
        if (endpoints.size() == 0) {
            httpHosts[0] = HttpHost.create("127.0.0.1:9200");
            return httpHosts;
        }
        for (int i = 0; i < endpoints.size(); i++) {
            httpHosts[i] = HttpHost.create(schema.concat("://").concat(endpoints.get(i)));
        }
        return httpHosts;
    }

    /**
     * 创建HttpHost
     *
     * @param endpoints 地址集合
     * @return HttpHost
     */
    private static HttpHost[] makeHttpHost(@NonNull List<String> endpoints) {
        HttpHost[] httpHosts = new HttpHost[endpoints.size()];
        if (endpoints.size() == 0) {
            httpHosts[0] = HttpHost.create("127.0.0.1:9200");
            return httpHosts;
        }
        for (int i = 0; i < endpoints.size(); i++) {
            httpHosts[i] = HttpHost.create(endpoints.get(i));
        }
        return httpHosts;
    }

}
